package org.lisy.java.thread;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.ForkJoinWorkerThread;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.ForkJoinPool.ForkJoinWorkerThreadFactory;

/**
 * ForkJoinPool 线程池操作
 *  - 是ExecutorSerice的一个补充
 *  - 特别适合用于-分而治之，递归计算的算法
 * @author lisy
 */
public class ForkJoinPoolOperate {

	/**
	 * 主要方法:
	 *   fork(): 类似于线程的 Thread.start() 方法，但是它不是真的启动一个线程，而是将任务放入到工作队列中
	 *   join(): 类似于线程的 Thread.join() 方法，但是它不是简单地阻塞线程，而是利用工作线程运行其它任务。
	 * 当一个工作线程中调用了 join() 方法，它将处理其它任务，直到注意到目标子任务已经完成了
	 *   注意: join 方法阻塞等待结果，所以先都调用 fork 方法后，再调用 join
	 */
	
	/**
	 * 三个任务子类
	 *   RecursiveAction: 无返回值任务
	 *   RecursiveTask: 有返回值任务
	 *   CountedCompleter: 无返回值任务，完成任务后可以触发回调
	 */
	
	public static void main(String[] args) {
		
		// 应用举例:求和计算
		sumExample();
	}

	private static void sumExample() {
		// 构造数据
        int length = 100000000;
        long[] arr = new long[length];
        for (int i = 0; i < length; i++) {
            arr[i] = ThreadLocalRandom.current().nextInt(Integer.MAX_VALUE);
        }
        
        // 使用 ForkJoinPool 线程池
        forkJoinSum(arr);
        
        // CPU 核心数
		int availableProcessors = Runtime.getRuntime().availableProcessors();
		System.out.println("CPU cores: " + availableProcessors);
	}
	
	private static void forkJoinSum(long[] arr) {
        long start = System.currentTimeMillis();
        
        /**
         * parallelism: 并行度。默认情况下跟我们机器的 CPU 核心数保持一致，使用 Runtime.getRuntime().availableProcessors() 可以得到我们机器运行时可用的 CPU 核心数
         * factory: 创建新线程的工厂。默认情况下使用 ForkJoinWorkerThreadFactory defaultForkJoinWorkerThreadFactory
         * handler: 线程异常情况下的处理器。在线程执行任务时对由于某些无法预料的错误而导致任务线程中断时，该处理器会进行一些处理，默认情况为 null
         * asyncMode: 在ForkJoinPool中，每一个工作线程都有一个独立的任务队列，asyncMode 表示工作线程内的任务队列是采用何种方式进行调度，可以是先进先出FIFO，也可以是后进先出 LIFO。如果为 true，则线程池中的工作线程则使用 先进先出方式 进行任务调度，默认情况下是false 也就是默认为 LIFO 后进先出
         */
        // 提供公用创建方式: ForkJoinPool.commonPool()
        ForkJoinPool forkJoinPool = new ForkJoinPool(8, new MyForkJoinWorkerThreadFactory(), null, false);
        
        // 提交任务
        ForkJoinTask<Long> forkJoinTask = forkJoinPool.submit(new SumTask(arr, 0, arr.length));
		try {
			// 获取结果
			Long sum = forkJoinTask.get();
			System.out.println("sum: " + sum);
			System.out.println("fork join elapse: " + (System.currentTimeMillis() - start));
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ExecutionException e) {
			e.printStackTrace();
		} finally {
			forkJoinPool.shutdown();
		}
    }

    private static class SumTask extends RecursiveTask<Long> {
		private static final long serialVersionUID = 1L;
		
		private long[] arr;
        private int from;
        private int to;

        public SumTask(long[] arr, int from, int to) {
            this.arr = arr;
            this.from = from;
            this.to = to;
        }

        @Override
        protected Long compute() {
        	if (Thread.currentThread() instanceof MyWorkerThread) {
        		MyWorkerThread thread = (MyWorkerThread) Thread.currentThread();
            	thread.addTask();
        	}
            // 小于1000的时候直接相加，可灵活调整
            if (to - from <= 1000) {
                long sum = 0;
                for (int i = from; i < to; i++) {
                    // 模拟耗时
                    sum += (arr[i]/3*3/3*3/3*3/3*3/3*3);
                }
                return sum;
            }

            // 分成两段任务
            int middle = (from + to) / 2;
            SumTask left = new SumTask(arr, from, middle);
            SumTask right = new SumTask(arr, middle, to);

            // 提交左边的任务
            left.fork();
            // 右边的任务直接利用当前线程计算，节约开销 right.fork(); right.join();
            Long rightResult = right.compute();
            // 等待左边计算完毕
            Long leftResult = left.join();
            // 返回结果
            return leftResult + rightResult;
        }
    }
	
}

/**
 * 线程工厂
 */
class MyForkJoinWorkerThreadFactory implements ForkJoinWorkerThreadFactory {
	
	@Override
	public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
		return new MyWorkerThread(pool);
	}
}

/**
 * 自定义线程
 */
class MyWorkerThread extends ForkJoinWorkerThread {
	
	private static ThreadLocal<Integer> taskCounter = new ThreadLocal<Integer>();
	
	protected MyWorkerThread(ForkJoinPool pool) {
		super(pool);
	}
	
	@Override
    protected void onStart() {
        super.onStart();
        taskCounter.set(0);
    }
	
    @Override
    protected void onTermination(Throwable exception) {
        super.onTermination(exception);
    }
    
    public void addTask() {
        int counter = taskCounter.get().intValue();
        counter++;
        taskCounter.set(counter);
        System.out.println("thread name: " + getName() + " count: " + taskCounter.get());
    }
}
